package com.idol.rabbitmq.config;

import org.springframework.amqp.core.*;
import org.springframework.amqp.rabbit.connection.ConnectionFactory;
import org.springframework.amqp.rabbit.core.RabbitAdmin;
import org.springframework.amqp.rabbit.listener.DirectMessageListenerContainer;
import org.springframework.amqp.support.converter.Jackson2JsonMessageConverter;
import org.springframework.amqp.support.converter.MessageConverter;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;

import java.util.HashMap;
import java.util.Map;

/**
 * @author SupremeSir
 * @description
 * @className RabbitConfig
 * @date 2021/4/14 23:32
 **/
@Configuration
public class RabbitConfig {

    @Bean(name = "rabbitMessageListenerContainer")
    public DirectMessageListenerContainer listenerContainer(ConnectionFactory connectionFactory) {
        //写的时候，默认使用DEFAULT_NUM_THREADS =Runtime.getRuntime().availableProcessors() * 2个线程
        DirectMessageListenerContainer container = new DirectMessageListenerContainer(connectionFactory);
        container.setAcknowledgeMode(AcknowledgeMode.MANUAL);
        container.setPrefetchCount(10);
        // 设置每个队列可以有5个消费者
        container.setConsumersPerQueue(5);
        // 一次确认几个消息
        container.setMessagesPerAck(1);

        ThreadPoolTaskExecutor taskExecutor = new ThreadPoolTaskExecutor();
        taskExecutor.setCorePoolSize(10);
        taskExecutor.setMaxPoolSize(20);
        // 设置该属性，灵活设置并发
        container.setTaskExecutor(taskExecutor);

        return container;
    }

    /**
     * 添加消息转换器
     */
    @Bean
    public MessageConverter messageConverter() {
        return new Jackson2JsonMessageConverter();
    }

    @Bean
    public RabbitAdmin rabbitAdmin(ConnectionFactory connectionFactory) {
        return new RabbitAdmin(connectionFactory);
    }

    /**
     * 订单消息队列
     */
    @Bean
    public Queue orderQueue() {
        return QueueBuilder.durable("q.order").build();
    }

    /**
     * 延迟消息队列
     */
    @Bean
    public Queue ttlQueue() {
        Map<String, Object> args = new HashMap<>();
        // 10s消息到期
        args.put("x-message-ttl", 10000);
        args.put("x-dead-letter-exchange", "x.dlx");
        args.put("x-dead-letter-routing-key", "k.dlx");
        Queue queue = new Queue("q.ttl", true, false, false, args);
        return queue;
    }

    /**
     * 死信队列，用于取消用户订单
     */
    @Bean
    public Queue dlxQueue() {
        Queue dlq = new Queue("q.dlx", true, false, false, null);
        return dlq;
    }

    /**
     * 订单交换器
     */
    @Bean
    public Exchange orderExchange() {
        DirectExchange exchange = new DirectExchange("x.order", true, false, null);
        return exchange;
    }

    /**
     * 延迟队列交换器
     */
    @Bean
    public Exchange ttlExchange() {
        return new DirectExchange("x.ttl", true, false, null);
    }

    /**
     * 死信队列交换器
     */
    @Bean
    public Exchange dlxExchange() {
        DirectExchange exchange = new DirectExchange("x.dlx", true, false, null);
        return exchange;
    }

    /**
     * 用于发送下单，做分布式事务的MQ
     */
    @Bean
    public Binding orderBinding() {
        return BindingBuilder.bind(orderQueue()).to(orderExchange()).with("k.order").noargs();
    }

    /**
     * 用于等待用户支付的延迟队列绑定
     */
    @Bean
    public Binding ttlBinding() {
        return BindingBuilder.bind(ttlQueue()).to(ttlExchange()).with("k.ttl").noargs();
    }

    /**
     * 用于支付超时取消用户订单的死信队列绑定
     */
    @Bean
    public Binding dlxBinding() {
        return BindingBuilder.bind(dlxQueue()).to(dlxExchange()).with("k.dlx").noargs();
    }



}
